package com.shortUrl.starter.verticle;

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;

import com.shortUrl.starter.util.PrefixUtils;
import com.shortUrl.starter.util.RedisUtils;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.redis.client.RedisAPI;
import io.vertx.redis.client.RedisConnection;
import io.vertx.redis.client.Response;
import io.vertx.redis.client.ResponseType;

public class RedisVerticle extends AbstractVerticle {

    private static final int MAX_RECONNECT_RETRIES = 16;
    private RedisConnection client;
    private RedisAPI redis;

    public static final String SHORTURL_CREATE_ID_ADDRESS = "redis.shorturl.createId";
    public static final String SHORTURL_GET_ADDRESS = "redis.shorturl.get";
    public static final String SHORTURL_LIST_ADDRESS = "redis.shorturl.list";

    @Override
    public void start() throws Exception {

        createRedisClient(onCreate -> {
            if (onCreate.succeeded()) {
                System.out.println("Redis 连接成功！");
                redis = RedisAPI.api(client);
            } else if (onCreate.failed()) {
                System.out.println("Redis 连接失败！");
            }
        });

        createCURDService();

    }
    

    @Override
    public void stop() throws Exception {
        client.close();
    }


    private void createRedisClient(Handler<AsyncResult<RedisConnection>> handler) {

        RedisUtils.getRedisClient(vertx).connect(connectHadler -> {
            if (connectHadler.succeeded()) {
                client = connectHadler.result();
                client.exceptionHandler(e -> {
                    attemptReconnect(0);
                });
            }
            handler.handle(connectHadler);
        });
    }

    private void attemptReconnect(int retry) {
        System.out.println("第" + retry + "次尝试重连 Redis");
   
        if (retry > MAX_RECONNECT_RETRIES) {
        } else {
            long backoff = (long) (Math.pow(2, Math.min(retry, 10)) * 10);
            vertx.setTimer(backoff, timer -> createRedisClient(onReconnect -> {
                if (onReconnect.failed()) {
                    attemptReconnect(retry + 1);
                }
            }));
         }
    }

    private void createCURDService() {
        // 获取事件总线
        EventBus bus = vertx.eventBus();
        // 消息消费者
        MessageConsumer<JsonObject> consumner = bus.consumer("createId");
        consumner.handler(msg -> {
            // 处理器的处理体
            JsonObject json = msg.body();
            // 获取到原url
            String prefix = json.getString("prefix");
            String idPrefix = PrefixUtils.getYMDHPrefix();
            String redisPrefix = prefix + idPrefix;
            redis.incr(redisPrefix, handler -> {
                Response result = handler.result();
                String re = idPrefix + String.format("%1$06d", result.toInteger());
                // new JsonObject()
                msg.reply(re);
            });
            redis.expire(Arrays.asList(redisPrefix,"1800"));
        });

        bus.<JsonObject>consumer("cashShortUrl", msg -> {
            JsonObject shortUrl = msg.body();
            redis.setex(shortUrl.getString("shortUrl"), "1800", shortUrl.getString("longUrl")).onSuccess(handler -> {
                msg.reply("成功");
            });
        });

        bus.<String>consumer("getLongUrl", msg -> {
            String shortUrl = msg.body();
            System.out.println(shortUrl);
            redis.get(shortUrl).onSuccess(resultHandle -> {
                System.out.println(resultHandle);
                if (resultHandle == null) {
                    msg.fail(1, "不存在值");
                } else {
                    String result = resultHandle.toString();
                    msg.reply(result);
                }

            });
            
            redis.expire(Arrays.asList(shortUrl,"1800"));
        });

        bus.<String>consumer(SHORTURL_LIST_ADDRESS, msg -> {
            System.out.println("消息");

            JsonArray array = new JsonArray();
            redis.keys("*", res -> {
                if (res.succeeded()) {

                    if (res.result().type() == ResponseType.MULTI) {

                        AtomicInteger hasGetCount = new AtomicInteger();
                        final int size = res.result().size();

                        for (Response r : res.result()) {

                            redis.hgetall(r.toString(), res2 -> {
                                Response data = res2.result();
                                JsonObject json = new JsonObject();
                                json.put("shortUrl", r.toString());
                                array.add(json);

                                if (hasGetCount.incrementAndGet() == size) {
                                    msg.reply(array);
                                    System.out.println("list 请求 redis 已回复");
                                }
                            });
                        }
                    }
                }
            });

        });
    }

}
